package org.lsda.caller;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.lsda.job.*;
import org.lsda.util.*;

import java.util.*;

/**
 * This class is used to send the job request and ask for computation.
 * 
 * @author Chunqiu Zeng
 * 
 */
public class LsdaSender implements MessageListener, ExceptionListener {

	private static Log log = LogFactory.getLog(LsdaSender.class);

	/**
	 * This is used to denote the message queue for computing job
	 */
	private String queueName = Util.getBRQueName();

	/**
	 * url for the message broker
	 */
	private String url = Util.getBRUrl();

	/**
	 * user name for accessing the message broker
	 */
	private String userName = Util.getBRUser();

	/**
	 * password for accessing the message broker
	 */
	private String password = Util.getBRPasswd();

	/**
	 * whether to support the transaction
	 */
	private boolean transacted = true;

	/**
	 * used to store all the handlers for message
	 * */
	private ArrayList<ActionHandler> actionHandlers = new ArrayList<ActionHandler>();

	/**
	 * defualt constructor
	 * */
	public LsdaSender() {
	}

	/**
	 * constructor
	 * */
	public LsdaSender(LsdaAction action) throws Exception {
	}

	/**
	 * get the queue name for the broker
	 * */
	public String getQueueName() {
		return queueName;
	}

	/**
	 * set the queue name for the broker
	 * */
	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	/**
	 * get the url for the message broker
	 * */
	public String getUrl() {
		return url;
	}

	/**
	 * set the url for the message broker
	 * */
	public void setUrl(String url) {
		this.url = url;
	}

	/**
	 * get the user name 
	 * */
	public String getUserName() {
		return userName;
	}

	/**
	 * set user name 
	 * */
	public void setUserName(String userName) {
		this.userName = userName;
	}

	/**
	 * get the password
	 * */
	public String getPassword() {
		return password;
	}

	/**
	 * set the password
	 * */
	public void setPassword(String password) {
		this.password = password;
	}
	
	/**
	 * add action handler
	 * */
	public void addActionHandler(ActionHandler ah) {
		actionHandlers.add(ah);
	}
	
	/**
	 * clear all the action handler
	 * */
	public void clearActionHandler() {
		actionHandlers.clear();
	}

	/**
	 * mark action to start status
	 * */
	public void actionStart(LsdaAction action) throws Exception {
		java.sql.Connection sqlConn = null;
		try {
			sqlConn = Util.getConnection();
			Util.markActionStart(sqlConn, action.getSchedSeq(),
					action.getActionSeq());
		} catch (Exception ex) {
			throw ex;
		} finally {
			try {
				sqlConn.close();
			} catch (Throwable ignore) {

			}
		}

	}
	
	
	/**
	 * mark action to end status
	 * */
	public void actionEnd(LsdaAction action) throws Exception {
		java.sql.Connection sqlConn = null;
		try {
			sqlConn = Util.getConnection();
			Util.markActionEnd(sqlConn, action.getSchedSeq(),
					action.getActionSeq(), action.getStatus(), action.getInfo());
		} catch (Exception ex) {
			throw ex;
		} finally {
			try {
				sqlConn.close();
			} catch (Throwable ignore) {
			}
		}
	}
	
	/**
	 * send the action to call execution
	 * */
	public void send(LsdaAction action) {
		if (null == action)
			return;
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				userName, password, url);
		Connection jmsConn = null;

		try {
			jmsConn = connectionFactory.createConnection();
			jmsConn.start();
			Session session = jmsConn.createSession(transacted,
					Session.AUTO_ACKNOWLEDGE);

			Destination destination = session.createQueue(queueName);
			MessageProducer msgProd = session.createProducer(destination);

			Destination tempDest = session.createTemporaryQueue();
			MessageConsumer msgCons = session.createConsumer(tempDest);
			ObjectMessage msg = session.createObjectMessage(action);
			msg.setJMSReplyTo(tempDest);
			msg.setJMSCorrelationID("" + action.getSchedSeq());

			actionStart(action);// update the status in db

			log.debug("send the job request to message broker");
			msgProd.send(msg);
			msgProd.close();/////////
			session.commit();

			Message message = msgCons.receive();
			if (message instanceof ObjectMessage) {
				onMessage(message);
				session.commit();
			}
			msgCons.close();/////////
			session.close();////////
			jmsConn.stop();
			jmsConn.close();
			
		} catch (Exception ex) {
			log.error(ex.getMessage());
		}

	}

	/**
	 * call all the handlers to do action
	 * */
	public void doAct(LsdaAction act) {
		ActionHandler ah = null;
		for (int i = 0; i < actionHandlers.size(); i++) {
			ah = actionHandlers.get(i);
			ah.handle(act);
		}
		log.info("reply:" + act.getPrgName() + " " + act.getPrgParam() + " \n"
				+ act.getStatus() + " \n" + act.getInfo());
	}

	/**
	 * when message comes, the function will be called to handle the message
	 * */
	@Override
	public void onMessage(Message arg0) {
		// TODO Auto-generated method stub
		ObjectMessage om = (ObjectMessage) arg0;
		try {
			LsdaAction act = (LsdaAction) om.getObject();
			actionEnd(act);
			doAct(act);

		} catch (JMSException e) {
			// TODO Auto-generated catch block
			log.error(e.getMessage());
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * when exception occurs, this function will be called
	 * */
	@Override
	public void onException(JMSException arg0) {
		// TODO Auto-generated method stub
		log.error(arg0.getMessage());
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
	}

}
